/*
 * Copyright 2019 WeBank
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.webank.wedatasphere.linkis.entrance.restful;

import com.webank.wedatasphere.linkis.common.log.LogUtils;
import com.webank.wedatasphere.linkis.entrance.EntranceServer;
import com.webank.wedatasphere.linkis.entrance.annotation.EntranceServerBeanAnnotation;
import com.webank.wedatasphere.linkis.entrance.conf.EntranceConfiguration;
import com.webank.wedatasphere.linkis.entrance.execute.EntranceJob;
import com.webank.wedatasphere.linkis.entrance.log.LogReader;
import com.webank.wedatasphere.linkis.entrance.utils.JobHistoryHelper;
import com.webank.wedatasphere.linkis.governance.common.entity.job.JobRequest;
import com.webank.wedatasphere.linkis.protocol.constants.TaskConstant;
import com.webank.wedatasphere.linkis.protocol.engine.JobProgressInfo;
import com.webank.wedatasphere.linkis.protocol.utils.ZuulEntranceUtils;
import com.webank.wedatasphere.linkis.rpc.Sender;
import com.webank.wedatasphere.linkis.scheduler.listener.LogListener;
import com.webank.wedatasphere.linkis.scheduler.queue.Job;
import com.webank.wedatasphere.linkis.scheduler.queue.SchedulerEventState;
import com.webank.wedatasphere.linkis.server.Message;
import com.webank.wedatasphere.linkis.server.security.SecurityFilter;
import org.apache.commons.dbcp.BasicDataSource;
import org.apache.commons.dbcp.BasicDataSourceFactory;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.codehaus.jackson.JsonNode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import scala.Option;

import javax.servlet.http.HttpServletRequest;
import javax.ws.rs.*;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.*;

/**
 * Description: an implementation class of EntranceRestfulRemote
 */
@Path("/entrance")
@Component
@Produces(MediaType.APPLICATION_JSON)
@Consumes(MediaType.APPLICATION_JSON)
public class EntranceRestfulApi implements EntranceRestfulRemote {
    private static final Logger logger = LoggerFactory.getLogger(EntranceRestfulApi.class);
    private static String globalFileName;
    private static LinkedHashMap<String, LinkedHashMap> globalParamsMap;
    private static String[] globalEndTimeFields;
    private static boolean whetherIncrementalMode = false;

    private EntranceServer entranceServer;

    @EntranceServerBeanAnnotation.EntranceServerAutowiredAnnotation
    public void setEntranceServer(EntranceServer entranceServer) {
        this.entranceServer = entranceServer;
    }

    /**
     * The execute function handles the request submitted by the user to execute the task, and the execution ID is returned to the user.
     * execute函数处理的是用户提交执行任务的请求，返回给用户的是执行ID
     * json Incoming key-value pair(传入的键值对)
     * Repsonse
     */
    @Override
    @POST
    @Path("/execute")
    public Response execute(@Context HttpServletRequest req, Map<String, Object> json) {
        Message message = null;
        logger.info("Begin to get an execID");
        json.put(TaskConstant.UMUSER, SecurityFilter.getLoginUsername(req));
        HashMap<String, String> map = (HashMap) json.get(TaskConstant.SOURCE);
     /*   HashMap<String, Object> hashMap = (HashMap) json.get(TaskConstant.PARAMS);
        HashMap<String, Object> objectHashMap = (HashMap) hashMap.get(TaskConstant.PARAMS_CONFIGURATION);
        HashMap<String, String> stringStringHashMap = (HashMap) objectHashMap.get(TaskConstant.PARAMS_CONFIGURATION_RUNTIME);*/

        /*   boolean containsKey = stringStringHashMap.containsKey("jdbc.url");*/
     /*   if (!containsKey) {
            Map<String, String> mapList = null;
            try {
                mapList = configurationRestfulApi.getJdbcFullTreesByAppName(req, SecurityFilter.getLoginUsername(req), "jdbc", "4", "IDE");
            } catch (ConfigurationException e) {
                e.printStackTrace();
            }
            stringStringHashMap.put("jdbc.url", mapList != null ? mapList.get("jdbc.url") : null);
            stringStringHashMap.put("jdbc.username", mapList != null ? mapList.get("jdbc.username") : null);
            stringStringHashMap.put("jdbc.password", mapList != null ? mapList.get("jdbc.password") : null);
        }*/


        if (map == null) {
            map = new HashMap<>();
            json.put(TaskConstant.SOURCE, map);
        }
        //获取参数：是否增量，增量字段
        //LinkedHashMap cannot be cast to java.lang.String
        String startTimeField = "";
        String endTimeField = "";
        String autoIncrementField = "";
        LinkedHashMap<String, LinkedHashMap> configurationMap = new LinkedHashMap<>();
        LinkedHashMap<String, String> variableMap = new LinkedHashMap<>(16);
        try {
            LinkedHashMap<String, LinkedHashMap> paramsMap = (LinkedHashMap<String, LinkedHashMap>) json.get("params");
            globalParamsMap = paramsMap;
            //自定义参数
            variableMap = paramsMap.get("variable");
            configurationMap = paramsMap.get("configuration");
        } catch (Exception e) {
            logger.error("lrd->EntranceRestfulApi->execute 参数获取异常" + e);
        }
        try {
            startTimeField = variableMap.get("start_time").toLowerCase();
        } catch (NullPointerException e) {
            logger.warn("lrd->1-EntranceRestfulApi->execute参数‘起始时间字段’ 没有配置 key eg: start_time value eg:table1.field1 或者 table1.field1;table2.field2");
        }
        try {
            endTimeField = variableMap.get("end_time").toLowerCase();
        } catch (NullPointerException e) {
            logger.warn("lrd->2-EntranceRestfulApi->execute参数‘结束时间字段’ 没有配置 key eg: end_time value eg:table1.field1 或者 table1.field1;table2.field2");
        }
        try {
            autoIncrementField = variableMap.get("id").toLowerCase();
        } catch (NullPointerException e) {
            logger.warn("lrd->3-EntranceRestfulApi->execute参数‘自增id字段’ 没有配置 key eg:id value eg:table1.field1 或者 table1.field1;table2.field2 ");
        }
        if (!StringUtils.isBlank(startTimeField + endTimeField + autoIncrementField)) {
            whetherIncrementalMode = true;
        }
        if (whetherIncrementalMode) {
            //获取etl数据的jdbc配置
            LinkedHashMap<String, String> jdbcInfoMap = configurationMap.get("runtime");
            //通过配置文件，获取系统的jdbc配置
            Connection conn = null;
            BasicDataSource dataSource = null;
            Statement statement = null;
            ResultSet rs = null;
            ResultSet queryNodeRes = null;
            //k 自增的表名，v 自增的字段名
            String[] endTimeFields = new String[1];
            //结束时间
            HashMap<String, String> endTimeData = new HashMap<String, String>();
            //获取自增表名和字段名
            if (StringUtils.isBlank(endTimeField)) {
                logger.warn("lrd->4-EntranceRestfulApi->execute: end_time 参数没有配置，或者配的k不叫end_time");
            } else if (!endTimeField.contains(".")) {
                logger.warn("lrd->5-EntranceRestfulApi->execute: end_time 参数不正确 正确示例：表名.字段名 或者 表名1.字段名1;表名2.字段名2");
            } else if (endTimeField.contains(";")) {
                //多个表名字段名
                endTimeFields = endTimeField.split(";");
            } else {
                //一个表名字段名
                endTimeFields[0] = endTimeField;
            }
            globalEndTimeFields = endTimeFields;
            try {
                Properties properties = new Properties();
                String nodeName = jdbcInfoMap.get("nodeName");
                properties.put("driver", "com.mysql.cj.jdbc.Driver");
                properties.put("url", jdbcInfoMap.get("jdbc.url"));
                properties.put("username", jdbcInfoMap.get("jdbc.username"));
                properties.put("password", jdbcInfoMap.get("jdbc.password"));
                properties.put("maxIdle", 20);
                properties.put("minIdle", 0);
                dataSource = (BasicDataSource) BasicDataSourceFactory.createDataSource(properties);
                conn = dataSource.getConnection();
                statement = conn.createStatement();
                for (String tableNameField : endTimeFields) {
                    String tableName = "";
                    if (!StringUtils.isBlank(tableNameField) && tableNameField.contains(".")) {
                        tableName = tableNameField.substring(0, tableNameField.indexOf("."));
                    }
                    statement.execute("CREATE TABLE if not exists yunpi_job_schedule (\n" +
                            "  node_name varchar(200) ,\n" +
                            "  table_name varchar(200) ,\n" +
                            "  start_time varchar(200),\n" +
                            "  end_time varchar(200) ,\n" +
                            "  id int\n" +
                            ");");
                    //查询之前，先查询，是否有这个node_name，有则查，无则插入初始值
                    String queryNodeSql = String.format("select count(*) as node from yunpi_job_schedule where node_name = '%s' and table_name = '%s';", nodeName, tableName);
                    queryNodeRes = statement.executeQuery(queryNodeSql);
                    int node = 0;
                    while (queryNodeRes.next()) {
                        node = Integer.parseInt(queryNodeRes.getString("node"));
                    }
                    if (node == 0) {
                        String initializationSql = String.format("INSERT INTO yunpi_job_schedule (node_name,table_name,end_time) VALUES ('%s','%s','1997-11-11 11:00:00');", nodeName, tableName);
                        statement.execute(initializationSql);
                    }

                    String sql = String.format("select end_time from yunpi_job_schedule where node_name = '%s' and table_name = '%s';", nodeName, tableName);
                    logger.info("lrd->-6-EntranceRestfulApi->execute sql：" + sql);
                    rs = statement.executeQuery(sql);
                    while (rs.next()) {
                        String value = rs.getString("end_time");
                        endTimeData.put(tableNameField, value);
                    }
                }
            } catch (SQLException e) {
                logger.error("lrd->-7-EntranceRestfulApi->execute sql执行发生异常：" + e.getMessage());
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    assert rs != null;
                    rs.close();
                    statement.close();
                    conn.close();
                    dataSource.close();
                } catch (NullPointerException | SQLException e) {
                    logger.error("lrd->-8-EntranceRestfulApi->execute 数据库连接关闭异常：" + e);
                }
            }

            //拼接sql
            String fileName = map.get("scriptPath");
            globalFileName = fileName;
            String sqlCode = "";
            if (!StringUtils.isBlank(fileName) && fileName.endsWith(".jdbc")) {
                try {
                    sqlCode = (String) json.get("executionCode");
                    String newSql = sqlCode;
                    for (String tableNameField : endTimeData.keySet()) {
                        newSql = newSql.replace("${" + tableNameField + "}", "'" + endTimeData.get(tableNameField) + "'");
                    }
                    json.put("executionCode", newSql);
                } catch (Exception e) {
                    logger.error("lrd->-9-EntranceRestfulApi->execute替换sql失败，失败原因 " + e);
                }
            }
        }

        String ip = JobHistoryHelper.getRequestIpAddr(req);
        map.put(TaskConstant.REQUEST_IP, ip);
        String execID = entranceServer.execute(json);
        Job job = entranceServer.getJob(execID).get();
        JobRequest jobReq = ((EntranceJob) job).getJobRequest();
        Long taskID = jobReq.getId();
        pushLog(LogUtils.generateInfo("You have submitted a new job, script code (after variable substitution) is"), job);
        pushLog("************************************SCRIPT CODE************************************", job);
        pushLog(jobReq.getExecutionCode(), job);
        pushLog("************************************SCRIPT CODE************************************", job);
        pushLog(LogUtils.generateInfo("Your job is accepted,  jobID is " + execID + " and taskID is " + taskID + " in " + Sender.getThisServiceInstance().toString() + ". Please wait it to be scheduled"), job);
        execID = ZuulEntranceUtils.generateExecID(execID, Sender.getThisServiceInstance().getApplicationName(), new String[]{Sender.getThisInstance()});
        message = Message.ok();
        message.setMethod("/api/entrance/execute");
        message.data("execID", execID);
        message.data("taskID", taskID);
        logger.info("End to get an an execID: {}, taskID: {}", execID, taskID);
//        }catch(ErrorException e){
//            message = Message.error(e.getDesc());
//            message.setStatus(1);
//            message.setMethod("/api/entrance/execute");
//        }
        return Message.messageToResponse(message);

    }

    @Override
    @POST
    @Path("/submit")
    public Response submit(@Context HttpServletRequest req, Map<String, Object> json) {
        Message message = null;
        logger.info("Begin to get an execID");
        json.put(TaskConstant.SUBMIT_USER, SecurityFilter.getLoginUsername(req));
        HashMap<String, String> map = (HashMap) json.get(TaskConstant.SOURCE);
        if (map == null) {
            map = new HashMap<>();
            json.put(TaskConstant.SOURCE, map);
        }
        String ip = JobHistoryHelper.getRequestIpAddr(req);
        map.put(TaskConstant.REQUEST_IP, ip);
        String execID = entranceServer.execute(json);
        Job job = entranceServer.getJob(execID).get();
        JobRequest jobRequest = ((EntranceJob) job).getJobRequest();
        Long taskID = jobRequest.getId();
        pushLog(LogUtils.generateInfo("You have submitted a new job, script code (after variable substitution) is"), job);
        pushLog("************************************SCRIPT CODE************************************", job);
        pushLog(jobRequest.getExecutionCode(), job);
        pushLog("************************************SCRIPT CODE************************************", job);
        pushLog(LogUtils.generateInfo("Your job is accepted,  jobID is " + execID + " and taskID is " + taskID + " in " + Sender.getThisServiceInstance().toString() + ". Please wait it to be scheduled"), job);
        execID = ZuulEntranceUtils.generateExecID(execID, Sender.getThisServiceInstance().getApplicationName(), new String[]{Sender.getThisInstance()});
        message = Message.ok();
        message.setMethod("/api/entrance/submit");
        message.data("execID", execID);
        message.data("taskID", taskID);
        logger.info("End to get an an execID: {}, taskID: {}", execID, taskID);
        return Message.messageToResponse(message);
    }

    private void pushLog(String log, Job job) {
        entranceServer.getEntranceContext().getOrCreateLogManager().onLogUpdate(job, log);
    }

    @Override
    @GET
    @Path("/{id}/status")
    public Response status(@PathParam("id") String id, @QueryParam("taskID") String taskID) {
        Message message = null;
        String realId = ZuulEntranceUtils.parseExecID(id)[3];
        Option<Job> job = Option.apply(null);
        try {
            job = entranceServer.getJob(realId);
        } catch (Exception e) {
            logger.warn("获取任务 {} 状态时出现错误", realId, e.getMessage());
            long realTaskID = Long.parseLong(taskID);
            String status = JobHistoryHelper.getStatusByTaskID(realTaskID);
            message = Message.ok();
            message.setMethod("/api/entrance/" + id + "/status");
            message.data("status", status).data("execID", id);
            return Message.messageToResponse(message);
        }
        if (job.isDefined()) {
            message = Message.ok();
            message.setMethod("/api/entrance/" + id + "/status");
            message.data("status", job.get().getState().toString()).data("execID", id);
            String status = job.get().getState().toString();
            //lrd->:决定增量，操作类型是jdbc，且运行成功，进行更新sql
            if (whetherIncrementalMode && globalFileName.endsWith(".jdbc") && "Succeed".equals(status)) {
                //取jdbc配置
                LinkedHashMap<String, LinkedHashMap> configurationMap = globalParamsMap.get("configuration");
                LinkedHashMap<String, String> jdbcInfoMap = configurationMap.get("runtime");
                //通过配置文件，获取系统的jdbc配置
                Connection conn = null;
                BasicDataSource dataSource = null;
                Statement statement = null;
                ResultSet queryrs = null;
                boolean updaters = false;
                //这个runtime里面是jdbc配置
                try {
                    Properties properties = new Properties();
                    String nodeName = jdbcInfoMap.get("nodeName");
                    properties.put("driver", "com.mysql.cj.jdbc.Driver");
                    properties.put("url", jdbcInfoMap.get("jdbc.url"));
                    properties.put("username", jdbcInfoMap.get("jdbc.username"));
                    properties.put("password", jdbcInfoMap.get("jdbc.password"));
                    properties.put("maxIdle", 20);
                    properties.put("minIdle", 0);
                    dataSource = (BasicDataSource) BasicDataSourceFactory.createDataSource(properties);
                    conn = dataSource.getConnection();
                    statement = conn.createStatement();
                    for (String tableNameField : globalEndTimeFields) {
                        String tableName = "";
                        String field = "";
                        String endTime = "";
                        if (!StringUtils.isBlank(tableNameField) && tableNameField.contains(".")) {
                            tableName = tableNameField.substring(0, tableNameField.indexOf("."));
                            field = tableNameField.substring(tableNameField.indexOf(".") + 1);
                        }
                        String queryEndTimeSql = String.format("select max(%s) as end_time from %s;", field, tableName);
                        logger.info("lrd->-0-EntranceRestfulApi->status->queryEndTimeSql：" + queryEndTimeSql);
                        queryrs = statement.executeQuery(queryEndTimeSql);
                        while (queryrs.next()) {
                            endTime = queryrs.getString("end_time");
                        }
                        String updateEndTimeSql = String.format("update yunpi_job_schedule set end_time = '%s' where node_name = '%s' and table_name = '%s'", endTime, nodeName, tableName);
                        logger.info("lrd->-1-EntranceRestfulApi->status->updateEndTimeSql：" + updateEndTimeSql);
                        updaters = statement.execute(updateEndTimeSql);
                        if (updaters) {
                            logger.error("lrd->-2-EntranceRestfulApi->status更新增量表失败！");
                        }
                    }
                } catch (SQLException e) {
                    logger.error("lrd->-3-EntranceRestfulApi->status sql执行发生异常：" + e.getMessage());
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    try {
                        assert queryrs != null;
                        queryrs.close();
                        statement.close();
                        conn.close();
                        dataSource.close();
                    } catch (NullPointerException | SQLException e) {
                        logger.error("lrd->-4-EntranceRestfulApi->status 数据库连接关闭异常：" + e);
                    }
                }
            }
        } else {
            message = Message.error("ID The corresponding job is empty and cannot obtain the corresponding task status.(ID 对应的job为空，不能获取相应的任务状态)");
        }
        return Message.messageToResponse(message);
    }

    @Override
    @GET
    @Path("/{id}/progress")
    public Response progress(@PathParam("id") String id) {
        Message message = null;
        String realId = ZuulEntranceUtils.parseExecID(id)[3];
        Option<Job> job = entranceServer.getJob(realId);
        if (job.isDefined()) {
            JobProgressInfo[] jobProgressInfos = ((EntranceJob) job.get()).getProgressInfo();
            if (jobProgressInfos == null) {
                message = Message.error("Can not get the corresponding progress information, it may be that the corresponding progress information has not been generated(不能获取相应的进度信息,可能是相应的进度信息还未生成)");
                message.setMethod("/api/entrance/" + id + "/progress");
            } else {
                List<Map<String, Object>> list = new ArrayList<>();
                for (JobProgressInfo jobProgressInfo : jobProgressInfos) {
                    if ("true".equals(EntranceConfiguration.PROGRESS_PUSH().getValue())) {
                        Map<String, Object> map = new HashMap<>();
                        map.put("id", jobProgressInfo.id());
                        map.put("succeedTasks", jobProgressInfo.succeedTasks());
                        map.put("failedTasks", jobProgressInfo.failedTasks());
                        map.put("runningTasks", jobProgressInfo.runningTasks());
                        map.put("totalTasks", jobProgressInfo.totalTasks());
                        list.add(map);
                    } else if (jobProgressInfo.failedTasks() > 0 || jobProgressInfo.runningTasks() > 0) {
                        Map<String, Object> map = new HashMap<>();
                        map.put("id", jobProgressInfo.id());
                        map.put("succeedTasks", jobProgressInfo.succeedTasks());
                        map.put("failedTasks", jobProgressInfo.failedTasks());
                        map.put("runningTasks", jobProgressInfo.runningTasks());
                        map.put("totalTasks", jobProgressInfo.totalTasks());
                        list.add(map);
                    }
                }
                message = Message.ok();
                message.setMethod("/api/entrance/" + id + "/progress");
                //TODO 去掉绝对值判断
                message.data("progress", Math.abs(job.get().getProgress())).data("execID", id).data("progressInfo", list);
            }
        } else {
            message = Message.error("The job corresponding to the ID is empty, and the corresponding task progress cannot be obtained.(ID 对应的job为空，不能获取相应的任务进度)");
        }
        return Message.messageToResponse(message);
    }

    @Override
    @GET
    @Path("/{id}/log")
    public Response log(@Context HttpServletRequest req, @PathParam("id") String id) {
        String realId = ZuulEntranceUtils.parseExecID(id)[3];
        Option<Job> job = Option.apply(null);
        Message message = null;
        try {
            job = entranceServer.getJob(realId);
        } catch (final Throwable t) {
            message = Message.error("The job you just executed has ended. This interface no longer provides a query. It is recommended that you download the log file for viewing.(您刚刚执行的job已经结束，本接口不再提供查询，建议您下载日志文件进行查看)");
            message.setMethod("/api/entrance/" + id + "/log");
            return Message.messageToResponse(message);
        }
        if (job.isDefined()) {
            logger.debug("begin to get log for {}(开始获取 {} 的日志)", job.get().getId(), job.get().getId());
            LogReader logReader = entranceServer.getEntranceContext().getOrCreateLogManager().getLogReader(realId);
            int fromLine = 0;
            int size = 100;
            boolean distinctLevel = true;
            if (req != null) {
                String fromLineStr = req.getParameter("fromLine");
                String sizeStr = req.getParameter("size");
                if (StringUtils.isNotBlank(fromLineStr)) {
                    fromLine = Math.max(Integer.parseInt(fromLineStr), 0);
                }
                if (StringUtils.isNotBlank(sizeStr)) {
                    size = Integer.parseInt(sizeStr) >= 0 ? Integer.parseInt(sizeStr) : 10000;
                }
                String distinctLevelStr = req.getParameter("distinctLevel");
                if ("false".equals(distinctLevelStr)) {
                    distinctLevel = false;
                }
            }

            Object retLog = null;
            int retFromLine = 0;
            try {
                if (distinctLevel) {
                    String[] logs = new String[4];
                    retFromLine = logReader.readArray(logs, fromLine, size);
                    retLog = new ArrayList<String>(Arrays.asList(logs));
                } else {
                    StringBuilder sb = new StringBuilder();
                    retFromLine = logReader.read(sb, fromLine, size);
                    retLog = sb.toString();
                }
            } catch (IllegalStateException e) {
                logger.debug("Failed to get log information for :{}(为 {} 获取日志失败)", job.get().getId(), job.get().getId(), e);
                message = Message.ok();
                message.setMethod("/api/entrance/" + id + "/log");
                message.data("log", "").data("execID", id).data("fromLine", retFromLine + fromLine);
            } catch (final IllegalArgumentException e) {
                logger.debug("Failed to get log information for :{}(为 {} 获取日志失败)", job.get().getId(), job.get().getId(), e);
                message = Message.ok();
                message.setMethod("/api/entrance/" + id + "/log");
                message.data("log", "").data("execID", id).data("fromLine", retFromLine + fromLine);
                return Message.messageToResponse(message);
            } catch (final Exception e1) {
                logger.debug("Failed to get log information for :{}(为 {} 获取日志失败)", job.get().getId(), job.get().getId(), e1);
                message = Message.error("Failed to get log information(获取日志信息失败)");
                message.setMethod("/api/entrance/" + id + "/log");
                message.data("log", "").data("execID", id).data("fromLine", retFromLine + fromLine);
                return Message.messageToResponse(message);
            } finally {
                if (null != logReader && job.get().isCompleted()) {
                    IOUtils.closeQuietly(logReader);
                }
            }
            message = Message.ok();
            message.setMethod("/api/entrance/" + id + "/log");
            message.data("log", retLog).data("execID", id).data("fromLine", retFromLine + fromLine);
            logger.debug("success to get log for {} (获取 {} 日志成功)", job.get().getId(), job.get().getId());
        } else {
            message = Message.error("Can't find execID(不能找到execID): " + id + "Corresponding job, can not get the corresponding log(对应的job，不能获得对应的日志)");
            message.setMethod("/api/entrance/" + id + "/log");
        }
        return Message.messageToResponse(message);
    }

    @Override
    @POST
    @Path("/{id}/killJobs")
    public Response killJobs(@Context HttpServletRequest req, JsonNode jsonNode, @PathParam("id") String strongExecId) {
        JsonNode idNode = jsonNode.get("idList");
        JsonNode taskIDNode = jsonNode.get("taskIDList");
        ArrayList<Long> waitToForceKill = new ArrayList<>();
        if (idNode.size() != taskIDNode.size()) {
            return Message.messageToResponse(Message.error("The length of the ID list does not match the length of the TASKID list(id列表的长度与taskId列表的长度不一致)"));
        }
        if (!idNode.isArray() || !taskIDNode.isArray()) {
            return Message.messageToResponse(Message.error("Request parameter error, please use array(请求参数错误，请使用数组)"));
        }
        ArrayList<Message> messages = new ArrayList<>();
        for (int i = 0; i < idNode.size(); i++) {
            String id = idNode.get(i).asText();
            Long taskID = taskIDNode.get(i).asLong();
            String realId = ZuulEntranceUtils.parseExecID(id)[3];
            //通过jobid获取job,可能会由于job找不到而导致有looparray的报错,一旦报错的话，就可以将该任务直接置为Cancenlled
            Option<Job> job = Option.apply(null);
            try {
                job = entranceServer.getJob(realId);
            } catch (Exception e) {
                logger.warn("can not find a job in entranceServer, will force to kill it", e.getMessage());
                //如果在内存中找不到该任务，那么该任务可能已经完成了，或者就是重启导致的
                waitToForceKill.add(taskID);
                Message message = Message.ok("Forced Kill task (强制杀死任务)");
                message.setMethod("/api/entrance/" + id + "/kill");
                message.setStatus(0);
                messages.add(message);
                continue;
            }
            Message message = null;
            if (job.isEmpty()) {
                logger.warn("can not find a job in entranceServer, will force to kill it");
                waitToForceKill.add(taskID);
                message = Message.ok("Forced Kill task (强制杀死任务)");
                message.setMethod("/api/entrance/" + id + "/killJobs");
                message.setStatus(0);
                messages.add(message);
            } else {
                try {
                    logger.info("begin to kill job {} ", job.get().getId());
                    job.get().kill();
                    message = Message.ok("Successfully killed the job(成功kill了job)");
                    message.setMethod("/api/entrance/" + id + "/kill");
                    message.setStatus(0);
                    message.data("execID", id);
                    //ensure the job's state is cancelled in database
                    if (job.get() instanceof EntranceJob) {
                        EntranceJob entranceJob = (EntranceJob) job.get();
                        JobRequest jobReq = entranceJob.getJobRequest();
                        entranceJob.updateJobRequestStatus(SchedulerEventState.Cancelled().toString());
                        jobReq.setProgress("1.0f");
                        LogListener logListener = entranceJob.getLogListener().getOrElse(null);
                        if (null != logListener) {
                            logListener.onLogUpdate(entranceJob, "Job " + jobReq.getId() + " was kill by user successfully(任务" + jobReq.getId() + "已成功取消)");
                        }
                        this.entranceServer.getEntranceContext().getOrCreatePersistenceManager().createPersistenceEngine().updateIfNeeded(jobReq);
                    }
                    logger.info("end to kill job {} ", job.get().getId());
                } catch (Throwable t) {
                    logger.error("kill job {} failed ", job.get().getId(), t);
                    message = Message.error("An exception occurred while killing the job, kill failed(kill job的时候出现了异常，kill失败)");
                    message.setMethod("/api/entrance/" + id + "/kill");
                    message.setStatus(1);
                }
            }
            messages.add(message);
        }
        if (!waitToForceKill.isEmpty()) {
            JobHistoryHelper.forceBatchKill(waitToForceKill);
        }
        return Message.messageToResponse(Message.ok("停止任务成功").data("messages", messages));
    }

    @Override
    @GET
    @Path("/{id}/kill")
    public Response kill(@PathParam("id") String id, @QueryParam("taskID") long taskID) {
        String realId = ZuulEntranceUtils.parseExecID(id)[3];
        //通过jobid获取job,可能会由于job找不到而导致有looparray的报错,一旦报错的话，就可以将该任务直接置为Cancenlled
        Option<Job> job = Option.apply(null);
        try {
            job = entranceServer.getJob(realId);
        } catch (Exception e) {
            logger.warn("can not find a job in entranceServer, will force to kill it", e);
            //如果在内存中找不到该任务，那么该任务可能已经完成了，或者就是重启导致的
            JobHistoryHelper.forceKill(taskID);
            Message message = Message.ok("Forced Kill task (强制杀死任务)");
            message.setMethod("/api/entrance/" + id + "/kill");
            message.setStatus(0);
            return Message.messageToResponse(message);
        }
        Message message = null;
        if (job.isEmpty()) {
            logger.warn("can not find a job in entranceServer, will force to kill it");
            //如果在内存中找不到该任务，那么该任务可能已经完成了，或者就是重启导致的
            JobHistoryHelper.forceKill(taskID);
            message = Message.ok("Forced Kill task (强制杀死任务)");
            message.setMethod("/api/entrance/" + id + "/kill");
            message.setStatus(0);
            return Message.messageToResponse(message);
        } else {
            try {
                logger.info("begin to kill job {} ", job.get().getId());
                job.get().kill();
                message = Message.ok("Successfully killed the job(成功kill了job)");
                message.setMethod("/api/entrance/" + id + "/kill");
                message.setStatus(0);
                message.data("execID", id);
                //ensure the job's state is cancelled in database
                if (job.get() instanceof EntranceJob) {
                    EntranceJob entranceJob = (EntranceJob) job.get();
                    JobRequest jobReq = entranceJob.getJobRequest();
                    entranceJob.updateJobRequestStatus(SchedulerEventState.Cancelled().toString());
                    this.entranceServer.getEntranceContext().getOrCreatePersistenceManager().createPersistenceEngine().updateIfNeeded(jobReq);
                }
                logger.info("end to kill job {} ", job.get().getId());
            } catch (Throwable t) {
                logger.error("kill job {} failed ", job.get().getId(), t);
                message = Message.error("An exception occurred while killing the job, kill failed(kill job的时候出现了异常，kill失败)");
                message.setMethod("/api/entrance/" + id + "/kill");
                message.setStatus(1);
            }
        }
        return Message.messageToResponse(message);
    }

    @Override
    @GET
    @Path("/{id}/pause")
    public Response pause(@PathParam("id") String id) {
        String realId = ZuulEntranceUtils.parseExecID(id)[3];
        Option<Job> job = entranceServer.getJob(realId);
        Message message = null;
        if (job.isEmpty()) {
            message = Message.error("can not find the job of exexID :" + id + " can not pause (不能找到execID: " + id + "对应的job，不能进行pause)");
            message.setMethod("/api/entrance/" + id + "/pause");
            message.setStatus(1);
        } else {
            try {
                //todo job pause 接口还未实现和给出
                //job.pause();
                logger.info("begin to pause job {} ", job.get().getId());
                message = Message.ok("success to pause job (成功pause了job)");
                message.setStatus(0);
                message.data("execID", id);
                message.setMethod("/api/entrance/" + id + "/pause");
                logger.info("end to pause job {} ", job.get().getId());
            } catch (Throwable t) {
                logger.info("pause job {} failed ", job.get().getId());
                message = Message.error("Abnormal when pausing job, pause failed(pause job的时候出现了异常，pause失败)");
                message.setMethod("/api/entrance/" + id + "/pause");
                message.setStatus(1);
            }
        }

        return Message.messageToResponse(message);
    }

}
